Add lag-based wait for future-version OFFLINE->STANDBY transition - #2976
Add lag-based wait for future-version OFFLINE->STANDBY transition#2976jingy-li wants to merge 1 commit into
Conversation
Implements a best-effort, progress-based leader election safeguard: when a replica transitions OFFLINE->STANDBY for a future version whose push is still in progress (not the current version, not an already-ready future version), the transition now polls local version-topic lag and waits for it to drop below a configurable threshold before proceeding, instead of immediately becoming STANDBY. This reduces the chance that a lagging replica gets elected LEADER shortly after. The wait is bounded by a timeout and fails open (proceeds as before) if the feature is disabled, the ingestion task/lag can't be resolved, or the timeout elapses, so push jobs are never blocked indefinitely. New configs (VeniceServerConfig / ConfigKeys): - server.future.version.standby.lag.check.enabled (default false) - server.future.version.standby.lag.threshold (default 0) - server.future.version.standby.lag.check.timeout.seconds (default 300) - server.future.version.standby.lag.check.poll.interval.seconds (default 10) Changes: - StoreIngestionTask#getLocalVersionTopicLag: exposes per-partition local VT lag via the existing measureLagWithCallToPubSub primitive. - AbstractPartitionStateModel#waitUntilFutureVersionLagAcceptable: new polling-wait loop. - LeaderFollowerPartitionStateModel#onBecomeStandbyFromOffline: hooks the new wait into the in-progress future-version branch. Tests: - Unit tests for VeniceServerConfig getters/defaults, getLocalVersionTopicLag, and the new wait method's enabled/disabled/polling/timeout/fail-open paths in LeaderFollowerPartitionStateModelTest. - New integration test FutureVersionStandbyLagCheckTest covering enabled (lag catches up), enabled with immediate timeout (fail-open), and disabled (baseline) scenarios against a real cluster. Co-authored-by: Copilot <[email protected]>
There was a problem hiding this comment.
Pull request overview
Adds an optional, best-effort safeguard during Helix OFFLINE→STANDBY for in-progress future versions by waiting until the replica’s local version-topic lag drops below a configurable threshold (bounded by timeout and fail-open), reducing the chance a lagging replica is elected LEADER too soon.
Changes:
- Introduces new server configs for enabling the lag check, threshold, timeout, and polling interval.
- Exposes per-partition local version-topic lag via
StoreIngestionTask#getLocalVersionTopicLag. - Hooks a new polling wait into OFFLINE→STANDBY for in-progress future versions, plus unit/integration coverage.
Reviewed changes
Copilot reviewed 9 out of 9 changed files in this pull request and generated 1 comment.
Show a summary per file
| File | Description |
|---|---|
| internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/helixrebalance/FutureVersionStandbyLagCheckTest.java | Adds integration coverage for enabled/disabled/timeout (fail-open) scenarios. |
| internal/venice-common/src/main/java/com/linkedin/venice/ConfigKeys.java | Adds config keys and documentation for the lag-based standby gating feature. |
| clients/da-vinci-client/src/test/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTaskTest.java | Adds unit test validating the new local VT lag accessor behavior. |
| clients/da-vinci-client/src/test/java/com/linkedin/davinci/helix/LeaderFollowerPartitionStateModelTest.java | Adds unit tests for enabled/disabled/polling/timeout/fail-open paths and correct routing for current/future/backup versions. |
| clients/da-vinci-client/src/test/java/com/linkedin/davinci/config/VeniceServerConfigTest.java | Adds tests for new config defaults and overrides. |
| clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTask.java | Adds getLocalVersionTopicLag using existing PubSub lag measurement. |
| clients/da-vinci-client/src/main/java/com/linkedin/davinci/helix/LeaderFollowerPartitionStateModel.java | Invokes lag-wait only for in-progress future versions when enabled. |
| clients/da-vinci-client/src/main/java/com/linkedin/davinci/helix/AbstractPartitionStateModel.java | Implements the polling wait loop based on local VT lag/threshold/timeout. |
| clients/da-vinci-client/src/main/java/com/linkedin/davinci/config/VeniceServerConfig.java | Wires new configs into VeniceServerConfig getters and fields. |
Suppressed comments (1)
clients/da-vinci-client/src/main/java/com/linkedin/davinci/helix/AbstractPartitionStateModel.java:467
Utils.sleepintentionally swallowsInterruptedExceptionand returns false when interrupted; in this loop the return value is ignored, so an interrupt can turn into a rapid re-loop (or prevent timely shutdown of the transition thread). Handle the boolean return and preserve the interrupt status, then fail open (proceed to STANDBY) to match the method’s best-effort semantics.
Utils.sleep(Math.min(pollIntervalMs, remainingMs));
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| long timeoutMs = TimeUnit.MINUTES.toMillis(serverConfig.getFutureVersionStandbyLagCheckTimeoutMinutes()); | ||
| long pollIntervalMs = TimeUnit.MINUTES.toMillis(serverConfig.getFutureVersionStandbyLagCheckPollIntervalMinutes()); | ||
| long deadlineMs = System.currentTimeMillis() + timeoutMs; |
| Utils.getReplicaId(message.getResourceName(), getPartition())); | ||
| if (isCurrentVersion || isFutureVersionReady) { | ||
| waitConsumptionCompleted(resourceName, notifier); | ||
| } else if (isFutureVersion && getStoreAndServerConfigs().isFutureVersionStandbyLagCheckEnabled()) { |
There was a problem hiding this comment.
Could we restrict this wait to an explicitly active future-version push, e.g. version.getStatus() == VersionStatus.STARTED, rather than using "future and not PUSHED/ONLINE" as the proxy?
isFutureVersionReady() is false for terminal states such as KILLED, ERROR, and ROLLED_BACK. Those states can overlap with an OFFLINE -> STANDBY callback because killing and Helix-resource cleanup are asynchronous: the transition may already be running or queued when the version is marked KILLED. In the normal kill path the ingestion task will eventually clear its PCS, so this wait should fail open on a later poll, but it can still occupy the transition worker until that poll; delayed kill or cleanup can extend this to the timeout.
It would also be useful to re-check the status while polling, since a push can be STARTED when the wait begins and become KILLED or ERROR afterward.
| serverConfig.getFutureVersionStandbyLagCheckTimeoutMinutes()); | ||
| return; | ||
| } | ||
| Utils.sleep(Math.min(pollIntervalMs, remainingMs)); |
There was a problem hiding this comment.
This loop synchronously sleeps on the Helix state-transition executor. With the defaults, each lagging partition can hold one worker for up to 120 minutes. VeniceServerConfig still defaults LEADER_FOLLOWER_STATE_TRANSITION_THREAD_POOL_STRATEGY to SINGLE_POOL_STRATEGY (20 workers), so enough concurrent future-version transitions could delay current or backup-version transitions as well.
What is the current rollout/adoption of DUAL_POOL_STRATEGY in production? Is enabling this lag check intended to be coupled with the dual-pool rollout? If some clusters still use the single pool, should we validate or reject that configuration combination, or otherwise avoid holding a transition worker while polling? Even with dual pool, it would be useful to understand expected concurrency versus the default future-version pool size.
Summary
Implements a best-effort, progress-based leader election safeguard: when a replica transitions OFFLINE->STANDBY for a future version whose push is still in progress (not the current version, not an already-ready future version), the transition now polls local version-topic lag and waits for it to drop below a configurable threshold before proceeding, instead of immediately becoming STANDBY. This reduces the chance that a lagging replica gets elected LEADER shortly after.
The wait is bounded by a timeout and fails open (proceeds as before) if the feature is disabled, the ingestion task/lag can't be resolved, or the timeout elapses, so push jobs are never blocked indefinitely.
New configs (VeniceServerConfig / ConfigKeys)
server.future.version.standby.lag.check.enabled(defaultfalse)server.future.version.standby.lag.threshold(default1000records)server.future.version.standby.lag.check.timeout.minutes(default120= 2 hours)server.future.version.standby.lag.check.poll.interval.minutes(default15)Changes
StoreIngestionTask#getLocalVersionTopicLag: exposes per-partition local VT lag via the existingmeasureLagWithCallToPubSubprimitive.AbstractPartitionStateModel#waitUntilFutureVersionLagAcceptable: new polling-wait loop.LeaderFollowerPartitionStateModel#onBecomeStandbyFromOffline: hooks the new wait into the in-progress future-version branch (correctly distinguishing future versions from backup versions).Tests
VeniceServerConfiggetters/defaults,getLocalVersionTopicLag, and the new wait method's enabled/disabled/polling/timeout/fail-open paths inLeaderFollowerPartitionStateModelTest.FutureVersionStandbyLagCheckTestcovering enabled (lag catches up), enabled with immediate timeout (fail-open), and disabled (baseline) scenarios against a real cluster.🤖 Generated with GitHub Copilot CLI